package com.atguigu.sparkcore.day04

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable

/**
 * Author atguigu
 * Date 2020/10/31 10:02
 */
class MapAcc extends AccumulatorV2[Double, mutable.Map[String, Double]] {
    
    private val _map = mutable.Map[String, Double]()
    
    // 判断map集合是空
    override def isZero: Boolean = _map.isEmpty
    
    // 复制:
    override def copy(): AccumulatorV2[Double, mutable.Map[String, Double]] = {
        val acc = new MapAcc
        acc._map ++= this._map
        acc
    }
    
    override def reset(): Unit = {
        _map.clear()
    }
    
    override def add(v: Double): Unit = {
        //  提供 和, 平均值, 个数, 最大值, 最小值
        _map("sum") = _map.getOrElse("sum", 0D) + v
        _map("count") = _map.getOrElse("count", 0D) + 1D
        _map("max") = _map.getOrElse("max", Double.MinValue).max(v)
        _map("min") = _map.getOrElse("min", Double.MaxValue).min(v)
    }
    
    override def merge(other: AccumulatorV2[Double, mutable.Map[String, Double]]): Unit = {
        // map合并: (聚合)
        /*other match {
            case o: MapAcc =>
                val om = o._map
                val tm = this._map
                // 把om的相关值合并到this的map
                om.foreach {
                    case ("sum", value) =>
                        tm("sum") = tm.getOrElse("sum", 0D) + value
                    case ("count", value) =>
                        tm("count") = tm.getOrElse("count", 0D) + value
                    case ("max", value) =>
                        tm("max") = tm.getOrElse("max", Double.MinValue).max(value)
                    case ("min", value) =>
                        tm("min") = tm.getOrElse("min", Double.MaxValue).min(value)
                }
            case _ =>
                throw new UnsupportedOperationException
        }*/
        
        other match {
             case o: MapAcc =>
                 val om = o._map
                 val tm = this._map
                 om.foldLeft(tm) {
                     case (map, ("sum", value)) =>
                         map("sum") = map.getOrElse("sum", 0D) + value
                         map
                     case (map, ("count", value)) =>
                         map("count") = map.getOrElse("count", 0D) + value
                         map
                     case (map, ("max", value)) =>
                         map("max") = map.getOrElse("max", Double.MinValue).max(value)
                         map
                     case (map, ("min", value)) =>
                         map("min") = map.getOrElse("min", Double.MaxValue).min(value)
                         map
                 }
             
             case _ =>
                 throw new UnsupportedOperationException
         }
        
    }
    
    override def value: mutable.Map[String, Double] = {
        _map("avg") = _map.getOrElse("sum", 0D) / _map.getOrElse("count", 1D)
        _map
    }
}

/*
value: 提供 和, 平均值, 个数, 最大值, 最小值
    Map["sum"-> , "avg" -> "count" -> , ...]

 */